# @Version        : 1.0
# @Update Time    : 2025/10/15 00:00
# @File           : auth_util.py
# @IDE            : PyCharm
# @Desc           : 认证工具类（门面模式实现）
from typing import Optional, Any, List, cast, Literal, Sequence

from fastapi import Response

from .auth_logic import AuthLogic
from .manager import AuthManager
from .parameter import LoginParameter, LogoutParameter
from ..models.session import TerminalInfoModel, AuthSessionModel
from ..models.token import TokenResponse, TokenInfo


class AuthUtil:
    """
    认证工具类
    提供简化的认证API，采用门面模式，隐藏AuthLogic的复杂性
    """

    # 默认登录类型
    DEFAULT_LOGIN_TYPE = "login"

    # 直接实例化AuthLogic对象
    _stp_logic: Optional[AuthLogic] = None

    @classmethod
    def get_login_type(cls) -> str:
        return cls._get_stp_logic().login_type

    @classmethod
    def set_login_type(cls, login_type: Optional[str] = None) -> None:
        """切换登录类型并绑定对应的 AuthLogic"""
        try:
            logic = AuthManager.get_logic(login_type)
        except Exception:
            logic = AuthLogic(login_type)
        cls._stp_logic = cast(AuthLogic, logic)

    @classmethod
    def set_stp_logic(cls, new_stp_logic: AuthLogic) -> None:
        """设置自定义的 AuthLogic 实例"""
        cls._stp_logic = new_stp_logic

    @classmethod
    def get_stp_logic(cls) -> AuthLogic:
        """获取当前使用的 AuthLogic 实例"""
        return cls._get_stp_logic()

    @classmethod
    def _get_stp_logic(cls) -> AuthLogic:
        """内部方法：优先返回已缓存逻辑，其次从 AuthManager 获取，最后本地创建"""
        if cls._stp_logic is not None:
            return cast(AuthLogic, cls._stp_logic)
        try:
            logic = AuthManager.get_logic(cls.DEFAULT_LOGIN_TYPE)
            cls._stp_logic = cast(AuthLogic, logic)
        except Exception:
            cls._stp_logic = AuthLogic(cls.DEFAULT_LOGIN_TYPE)
        return cast(AuthLogic, cls._stp_logic)

    # ------------------- Token 管理 -------------------

    @classmethod
    def get_token_name(cls) -> str:
        return cls._get_stp_logic().get_token_name()

    @classmethod
    def set_token_value(
        cls,
        response: Response,
        token_value: str,
        cookie_timeout: Optional[int] = None,
        login_parameter: Optional[LoginParameter] = None,
    ) -> None:
        cls._get_stp_logic().set_token_value(
            response, token_value, cookie_timeout, login_parameter
        )

    @classmethod
    def set_token_value_to_storage(cls, token_value: str) -> None:
        cls._get_stp_logic().set_token_value_to_storage(token_value)

    @classmethod
    def get_token_value(cls) -> str:
        return cls._get_stp_logic().get_token_value()

    @classmethod
    def get_token_value_not_cut(cls) -> str:
        return cls._get_stp_logic().get_token_value_not_cut()

    @classmethod
    async def get_token_info(cls) -> TokenInfo:
        return await cls._get_stp_logic().get_token_info()

    # ------------------- 登录相关操作 -------------------

    @classmethod
    async def login(
        cls,
        response: Response,
        login_id: Any,
        device_type: Optional[str] = None,
        is_lasting_cookie: Optional[bool] = None,
        timeout: Optional[int] = None,
        login_parameter: Optional[LoginParameter] = None,
    ) -> None:
        await cls._get_stp_logic().login(
            response=response,
            login_id=login_id,
            device_type=device_type,
            is_lasting_cookie=is_lasting_cookie,
            timeout=timeout,
            login_parameter=login_parameter,
        )

    @classmethod
    def result_ok(cls) -> TokenResponse:
        return cls._get_stp_logic().result_ok()

    # ------------------- 注销相关操作 -------------------

    @classmethod
    async def logout_by_token_value(
        cls, token_value: str, logout_parameter: Optional[LogoutParameter] = None
    ) -> None:
        await cls._get_stp_logic().logout_by_token_value(token_value, logout_parameter)

    @classmethod
    async def kickout(
        cls,
        login_id: Any,
        device_type: Optional[str] = None,
        logout_parameter: Optional[LogoutParameter] = None,
    ) -> None:
        await cls._get_stp_logic().kickout(
            login_id, device_type=device_type, logout_parameter=logout_parameter
        )

    @classmethod
    async def kickout_by_token_value(
        cls, token_value: str, logout_parameter: Optional[LogoutParameter] = None
    ) -> None:
        await cls._get_stp_logic().kickout_by_token_value(token_value, logout_parameter)

    @classmethod
    async def replaced_by_token_value(
        cls, token_value: str, logout_parameter: Optional[LogoutParameter] = None
    ) -> None:
        await cls._get_stp_logic().replaced_by_token_value(
            token_value, logout_parameter
        )

    @classmethod
    async def logout_by_login_id(
        cls,
        response: Response,
        *,  # 强制使用关键字参数
        login_id: Any,
        device_type: Optional[str] = None,
        logout_parameter: Optional[LogoutParameter] = None,
    ) -> None:
        await cls._get_stp_logic().logout(
            response,
            login_id=login_id,
            device_type=device_type,
            logout_parameter=logout_parameter,
        )

    @classmethod
    def remove_terminal_by_kickout(
        cls, session: AuthSessionModel, terminal: TerminalInfoModel
    ) -> None:
        cls._get_stp_logic().remove_terminal_by_kickout(session, terminal)

    @classmethod
    def remove_terminal_by_replaced(
        cls, session: AuthSessionModel, terminal: TerminalInfoModel
    ) -> None:
        cls._get_stp_logic().remove_terminal_by_replaced(session, terminal)

    @classmethod
    def remove_terminal_by_logout(
        cls, session: AuthSessionModel, terminal: TerminalInfoModel
    ) -> None:
        cls._get_stp_logic().remove_terminal_by_logout(session, terminal)

    # ------------------- 会话查询 -------------------

    @classmethod
    async def is_login(cls, login_id: Optional[Any] = None) -> bool:
        return await cls._get_stp_logic().is_login(login_id)

    @classmethod
    async def check_login(cls) -> None:
        await cls._get_stp_logic().check_login()

    @classmethod
    async def get_login_id(cls) -> Any:
        return await cls._get_stp_logic().get_login_id()

    @classmethod
    async def get_login_id_default_null(cls) -> Optional[Any]:
        return await cls._get_stp_logic().get_login_id_default_null()

    @classmethod
    async def get_login_id_as_string(cls) -> str:
        return str(await cls._get_stp_logic().get_login_id())

    @classmethod
    async def get_login_id_as_int(cls) -> int:
        return int(await cls._get_stp_logic().get_login_id())

    @classmethod
    async def get_login_id_by_token(cls, token_value: str) -> Optional[Any]:
        return await cls._get_stp_logic().get_login_id_by_token(token_value)

    @classmethod
    async def get_login_id_by_token_not_think_freeze(
        cls, token_value: str
    ) -> Optional[Any]:
        # 通过令牌获取登录ID，不要考虑冻结
        return await cls._get_stp_logic().get_login_id_by_token_not_think_freeze(
            token_value
        )

    # ------------------- Session 相关 -------------------

    @classmethod
    async def create_login_session(
        cls, login_id: Any, login_parameter: Optional[LoginParameter] = None
    ) -> str:
        return await cls._get_stp_logic().create_login_session(
            login_id, login_parameter
        )

    @classmethod
    async def get_or_create_login_session(cls, login_id: Any) -> str:
        """
        获取指定账号 id 的登录会话数据，如果获取不到则创建并返回

        Args:
            login_id: 账号id

        Returns:
            str: 会话令牌
        """
        return await cls._get_stp_logic().get_or_create_login_session(login_id)

    @classmethod
    async def get_session_by_login_id(
        cls, login_id: Any, is_create: bool = True
    ) -> AuthSessionModel:
        return await cls._get_stp_logic().get_session_by_login_id(login_id, is_create)

    @classmethod
    async def get_session_by_session_id(
        cls, session_id: str
    ) -> Optional[AuthSessionModel]:
        return await cls._get_stp_logic().get_session_by_session_id(session_id)

    @classmethod
    async def get_session(cls, is_create: Optional[bool] = None) -> AuthSessionModel:
        return await cls._get_stp_logic().get_session(is_create)

    @classmethod
    async def get_token_session_by_token(cls, token_value: str) -> AuthSessionModel:
        return await cls._get_stp_logic().get_token_session_by_token(token_value)

    @classmethod
    async def get_token_session(cls) -> AuthSessionModel:
        return await cls._get_stp_logic().get_token_session()

    @classmethod
    async def get_anon_token_session(cls) -> AuthSessionModel:
        return await cls._get_stp_logic().get_anon_token_session()

    # ------------------- 活跃度与过期时间 -------------------

    @classmethod
    async def update_last_active_to_now(cls) -> None:
        await cls._get_stp_logic().update_last_active_to_now()

    @classmethod
    async def check_active_timeout(cls) -> None:
        await cls._get_stp_logic().check_active_timeout()

    @classmethod
    async def get_token_last_active_time(cls) -> int:
        return await cls._get_stp_logic().get_token_last_active_time()

    @classmethod
    async def get_token_timeout(cls, token: Optional[str] = None) -> int:
        return await cls._get_stp_logic().get_token_timeout(token)

    @classmethod
    async def get_session_timeout(cls) -> int:
        """
        获取当前登录账号的 Account-Session 剩余有效时间（单位: 秒，返回 -1 代表永久有效，-2 代表没有这个值）
        """
        return await cls._get_stp_logic().get_session_timeout()

    @classmethod
    async def get_token_session_timeout(cls) -> int:
        """
        获取当前 token 的 Token-Session 剩余有效时间（单位: 秒，返回 -1 代表永久有效，-2 代表没有这个值）
        """
        return await cls._get_stp_logic().get_token_session_timeout()

    @classmethod
    async def get_token_active_timeout(cls) -> int:
        return await cls._get_stp_logic().get_token_active_timeout()

    @classmethod
    async def renew_timeout(
        cls, timeout: int, token_value: Optional[str] = None
    ) -> None:
        await cls._get_stp_logic().renew_timeout(timeout, token_value)

    # ------------------- 角色认证 -------------------

    @classmethod
    async def get_role_list(cls, login_id: Optional[Any] = None) -> Sequence[str]:
        return await cls._get_stp_logic().get_role_list(login_id)

    @classmethod
    async def has_single_role(cls, role: str, login_id: Optional[Any] = None) -> bool:
        return await cls._get_stp_logic().has_single_role(role, login_id)

    @classmethod
    async def check_role(cls, *role: str) -> None:
        await cls._get_stp_logic().check_role(*role)

    # ------------------- 权限认证 -------------------

    @classmethod
    async def get_permission_list(cls, login_id: Optional[Any] = None) -> Sequence[str]:
        return await cls._get_stp_logic().get_permission_list(login_id)

    @classmethod
    async def has_single_permission(
        cls, permission: str, login_id: Optional[Any] = None
    ) -> bool:
        return await cls._get_stp_logic().has_single_permission(permission, login_id)

    @classmethod
    async def has_permission(
        cls, *permission: str, logic: Literal["and", "or"] = "and"
    ) -> None:
        await cls._get_stp_logic().has_permission(*permission, logic=logic)

    @classmethod
    async def check_permission(
        cls, *permission: str, logic: Literal["and", "or"] = "and"
    ) -> None:
        await cls._get_stp_logic().check_permission(*permission, logic=logic)

    # ------------------- Token 反查 -------------------

    @classmethod
    async def get_token_value_by_login_id(
        cls, login_id: Any, device_type: Optional[str] = None
    ) -> Optional[str]:
        return await cls._get_stp_logic().get_token_value_by_login_id(
            login_id, device_type
        )

    @classmethod
    async def get_token_value_list_by_login_id(
        cls, login_id: Any, device_type: Optional[str] = None
    ) -> List[str]:
        return await cls._get_stp_logic().get_token_value_list_by_login_id(
            login_id, device_type
        )

    @classmethod
    async def get_terminal_list_by_login_id(
        cls, login_id: Any, device_type: Optional[str] = None
    ) -> List[TerminalInfoModel]:
        return await cls._get_stp_logic().get_terminal_list_by_login_id(
            login_id, device_type
        )

    @classmethod
    async def get_terminal_info(cls) -> Optional[TerminalInfoModel]:
        return await cls._get_stp_logic().get_terminal_info()

    @classmethod
    async def get_terminal_info_by_token(
        cls, token_value: str
    ) -> Optional[TerminalInfoModel]:
        return await cls._get_stp_logic().get_terminal_info_by_token(token_value)

    @classmethod
    async def get_login_device_type(cls) -> Optional[str]:
        return await cls._get_stp_logic().get_login_device_type()

    @classmethod
    async def get_login_device_type_by_token(cls, token_value: str) -> Optional[str]:
        return await cls._get_stp_logic().get_login_device_type_by_token(token_value)

    @classmethod
    async def get_login_device_id(cls) -> Optional[str]:
        return await cls._get_stp_logic().get_login_device_id()

    @classmethod
    async def get_login_device_id_by_token(cls, token_value: str) -> Optional[str]:
        return await cls._get_stp_logic().get_login_device_id_by_token(token_value)

    @classmethod
    async def is_trust_device_id(cls, login_id: Any, device_id: str) -> bool:
        return await cls._get_stp_logic().is_trust_device_id(login_id, device_id)

    # ------------------- 会话管理 -------------------

    @classmethod
    async def search_token_value(
        cls, keyword: str, start: int = 0, size: int = 10, sort_type: bool = True
    ) -> List[str]:
        return await cls._get_stp_logic().search_token_value(
            keyword, start, size, sort_type
        )

    @classmethod
    async def search_session_id(
        cls,
        keyword: Optional[str],
        start: int = 0,
        size: int = 10,
        sort_type: bool = True,
    ) -> List[str]:
        return await cls._get_stp_logic().search_session_id(
            keyword, start, size, sort_type
        )

    @classmethod
    async def search_token_session_id(
        cls,
        keyword: Optional[str],
        start: int = 0,
        size: int = 10,
        sort_type: bool = True,
    ) -> List[str]:
        return await cls._get_stp_logic().search_token_session_id(
            keyword, start, size, sort_type
        )

    # ------------------- 账号封禁 -------------------

    @classmethod
    async def disable(
        cls, login_id: Any, time: int, server: Optional[str] = None
    ) -> None:
        """
        此方法不会直接将此账号id踢下线，如需封禁后立即掉线，请追加调用 StpUtil.logout(id)
        Args:
            login_id: 指定账号id
            time: 封禁时间, 单位: 秒 （-1=永久封禁）
            server: 禁用服务

        Returns:

        """
        await cls._get_stp_logic().disable(login_id, time, server)

    @classmethod
    async def disable_by_token_value(
        cls,
        token_value: str,
        time: int = -1,
        logout_parameter: Optional[LogoutParameter] = None,
    ) -> Optional[Any]:
        """
        根据token值获取登录ID并封禁用户，同时可以选择是否立即下线

        Args:
            token_value: token值
            time: 封禁时间, 单位: 秒 （-1=永久封禁）
            logout_parameter: 注销参数

        Returns:
            Any: 被封禁的登录ID，如果token无效则返回None
        """
        login_id = await cls.get_login_id_by_token(token_value)
        if login_id is not None:
            await cls.disable(login_id, time)
            await cls.logout_by_token_value(token_value, logout_parameter)
        return login_id

    @classmethod
    async def is_disable(cls, login_id: Any) -> bool:
        """
        判断：指定账号是否已被封禁 (true=已被封禁, false=未被封禁)
        Args:
            login_id: 指定账号id

        Returns: bool

        """
        return await cls._get_stp_logic().is_disable(login_id)

    @classmethod
    async def check_disable(cls, login_id: Any, service: Optional[str] = None) -> None:
        """
        校验：指定账号是否已被封禁，如果被封禁则抛出异常
        Args:
            login_id:账号id
            service: 禁用服务

        Returns:

        """
        await cls._get_stp_logic().check_disable(login_id, service)

    @classmethod
    async def get_disable_time(cls, login_id: Any) -> int:
        return await cls._get_stp_logic().get_disable_time(login_id)

    @classmethod
    async def untie_disable(cls, login_id: Any) -> None:
        await cls._get_stp_logic().untie_disable(login_id)

    # ------------------- 工具方法 -------------------

    @classmethod
    def create_sa_login_parameter(cls) -> LoginParameter:
        return cls._get_stp_logic().create_login_parameter()

    @classmethod
    async def logout(
        cls,
        response: Response,
        *,  # 强制使用关键字参数
        login_id: Optional[Any] = None,
        device_type: Optional[str] = None,
    ) -> None:
        await cls._get_stp_logic().logout(
            response, login_id=login_id, device_type=device_type
        )
